import { PriorityQueue } from "../utils/queue/queue.priority";
import { logger } from "../configs/log.config";
import { QueueOverflowStrategy } from "../utils/queue/queue.enum";
import { Queue } from "../utils/queue/queue";
import { equals, is } from "ramda";
import { isProd, sleep, time2Millis } from "../utils/cmm.util";
import { CancelToken, CancelTokenSource } from "axios";
import { genCancelToken } from "../utils/request.util";
import { timer } from "rxjs";

/**
 * app 任务类
 * 1. 启动任务
 * 2. 暂停任务
 * 3. 停止任务
 */
export abstract class AppTask {
  /**
   * 构造函数
   * 添加任务ID
   */
  constructor(priority: number = 3) {
    this.__taskId = AppTask.__taskIdCounter();
    this.__priority = priority;
    this.addStopListener(this.__stop.bind(this));
    this.addStopListener(this.onStop.bind(this));
  }

  /**
   * 任务ID计数器
   * @private
   */
  private static __taskIdCounter = (() => {
    let tid = 0;
    return () => ++tid;
  })();

  // 任务ID
  private readonly __taskId: number;
  // 任务状态
  private __status: "running" | "stopped" | "paused" = "stopped";
  // 优先级
  private readonly __priority: number;

  // 任务是否已经启动过了
  private __started = false;

  // 当前活跃的任务数量
  private static activeCount = 0;

  /**
   * 获取当前活跃的任务数量
   */
  static getActiveCount = () => AppTask.activeCount;

  // 存储取消请求的token
  private __cancelSources: CancelTokenSource[] = [];

  /**
   * 添加取消请求的token
   * @param source
   */
  protected __addCancelToken(source?: CancelTokenSource): CancelToken {
    if (!source) source = genCancelToken();
    this.__cancelSources.push(source);
    return source.token;
  }

  /**
   * 取消请求
   * @protected
   */
  protected __cancelTokens() {
    this.__cancelSources.forEach((source) => source.cancel());
    this.__cancelSources = [];
  }

  /**
   * 获取任务是否停止
   */
  public isStopped() {
    return this.__status === "stopped";
  }

  // 获取任务是否运行
  public isRunning() {
    return this.__status === "running";
  }

  // 获取任务是否暂停
  public isPaused() {
    return this.__status === "paused";
  }

  /**
   * 获取任务ID
   */
  getTaskId() {
    return this.__taskId;
  }

  /**
   * 获取优先级
   */
  getPriority() {
    return this.__priority;
  }

  /**
   * 获取任务状态
   */
  getStatus() {
    return this.__status;
  }

  /**
   * 启动任务
   */
  async start(): Promise<void> {
    if (this.isRunning()) return;
    if (this.isPaused()) {
      await this.restore();
      return;
    }
    if (this.__started) return;
    this.__started = true;
    AppTask.activeCount++;
    this.__status = "running";
    try {
      await this.__run();
    } catch (e) {
      logger.error("任务执行异常", e);
    }
    await this.stop();
  }

  /**
   * 运行任务
   */
  abstract __run(): Promise<void>;

  /**
   * 暂停任务
   */
  async pause(): Promise<boolean> {
    if (!this.isRunning()) return false;
    try {
      const ans = await this.__pause();
      this.__status = "paused";
      return ans;
    } catch (e) {
      logger.error("任务暂停异常", e);
      return false;
    }
  }

  /**
   * 阻塞任务
   * @protected
   */
  protected async __await() {
    const interval = this.getTaskAwaitInterval();
    while (true) {
      if (this.isStopped()) break;
      await sleep(interval);
    }
  }

  /**
   * 获取阻塞任务检查间隔
   */
  getTaskAwaitInterval(): number {
    return time2Millis("1m");
  }

  /**
   * 暂停任务
   */
  abstract __pause(): Promise<boolean>;

  /**
   * 恢复任务
   */
  async restore(): Promise<boolean> {
    if (!this.isPaused()) return false;
    try {
      const ans = await this.__restore();
      this.__status = "running";
      return ans;
    } catch (e) {
      logger.error("任务恢复异常", e);
      return false;
    }
  }

  /**
   * 恢复任务
   */
  abstract __restore(): Promise<boolean>;

  /**
   * 停止任务
   */
  async stop(): Promise<boolean> {
    if (this.isStopped()) return false;
    this.__status = "stopped";
    AppTask.activeCount--;
    let ans = false;
    this.__stopWatchers.forEach((listener) => listener());
    this.__cancelSources.forEach((source) => source.cancel());
    return ans;
  }

  /**
   * 停止任务回调（用于覆写）；
   */
  onStop(): any {}

  /**
   * 停止任务
   */
  abstract __stop(): Promise<boolean>;

  private __stopWatchers: Array<() => void> = [];

  /**
   * 添加停止任务回调
   * @param listener
   */
  addStopListener(listener: () => void) {
    this.__stopWatchers.push(listener);
    return () => this.removeStopListener(listener);
  }

  /**
   * 添加停止任务回调
   * @param task
   */
  addStopListenerByTask(task: AppTask) {
    const stop = () => task.stop();
    this.__stopWatchers.push(stop);
    return () => this.removeStopListener(stop);
  }

  /**
   * 移除停止任务回调
   * @param listener
   */
  removeStopListener(listener: () => void) {
    this.__stopWatchers = this.__stopWatchers.filter(equals(listener));
  }
}

/**
 * 测试环境打印任务数量
 * 1. 主要是为了调试是否存在内存泄漏问题
 */
timer(1000).subscribe(() => {
  if (isProd() || 1) return;
  setInterval(() => {
    logger.info("task active count: %s", AppTask.getActiveCount());
  }, 3000);
});

////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////
/**
 * 任务队列
 * 1. 限制正在执行的任务数量
 * 2. 基于优先级队列
 */
export class AppTaskQueue {
  /**
   * 固定比较器，优先级高的先执行
   * @param a
   * @param b
   */
  private __comparer = (a: AppTask, b: AppTask) => {
    return b.getPriority() - a.getPriority();
  };

  // 优先级队列
  private __queue: Queue<AppTask>;

  /**
   * 获取当前队列长度
   */
  getQueueSize() {
    return this.__queue.size();
  }

  private __destroyed = false;

  /**
   * 构造函数
   * @param __workerCount 并发数量
   * @param queueOrMaxSize 队列最大长度 |
   * @param overflowStrategy 溢出策略， 默认丢弃头部
   */
  constructor(
    private __workerCount: number = 100,
    queueOrMaxSize: number | Queue<AppTask> = 0,
    overflowStrategy: QueueOverflowStrategy = QueueOverflowStrategy.DISCARD_HEAD,
  ) {
    if (__workerCount <= 0) logger.error("workerCount must be greater than 0 , %d", __workerCount);
    if (is(Number, queueOrMaxSize)) {
      this.__queue = new PriorityQueue<AppTask>(this.__comparer, queueOrMaxSize, overflowStrategy);
    } else {
      this.__queue = queueOrMaxSize;
    }
    this.__run().then();
  }

  /**
   * 添加任务
   * @param task
   */
  addTask(task: AppTask) {
    if (this.__destroyed) return;
    this.__queue.offer(task);
    this.__run().then();
  }

  /**
   * 添加任务
   * @param tasks
   */
  addTasks(tasks: AppTask[]) {
    tasks.forEach(this.addTask.bind(this));
  }

  /**
   * 运行任务
   */
  private async __run() {
    while (this.getRunningTaskCount() < this.__workerCount && !this.__queue.isEmpty()) {
      this.__runTask(this.__queue.poll()!).then();
    }
  }

  /**
   * 当前正在运行的任务
   * @private
   */
  private __runningTasks = new Set<AppTask>();

  /**
   * 运行单个任务
   * @param task 任务
   * @private
   */
  private async __runTask(task: AppTask) {
    if (!task) return;
    if (task.getStatus() === "running") return;
    this.__runningTasks.add(task);
    await task.start();
    this.__runningTasks.delete(task);
    this.__run().then();
  }

  /**
   * 获取正在运行的任务数量
   */
  getRunningTaskCount() {
    return this.__runningTasks.size;
  }

  /**
   * 清空任务
   * 1. 这里的停止任务，是异步的，管理器不会等待任务结束，仅仅调用任务的 stop 方法；
   */
  destroy() {
    this.__destroyed = true;
    for (const task of this.__runningTasks) {
      task.stop().then();
    }
    this.__queue.clear();
    this.__runningTasks.clear();
  }
}
